package com.seaboxdata.threadHandler.electronic;

import com.alibaba.fastjson.JSONObject;
import com.seaboxdata.entity.ElectronicCoupletUploadPicture;
import com.seaboxdata.kafka.BaseStringKafkaConsumer;
import com.seaboxdata.service.ElectronicCoupletUploadPictureService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * 水库接口1内置机KAFKA消费
 *
 * @Author wei
 * @Date 2021/8/5 16:15
 **/
@Component
@Slf4j
public class ElectronicCoupletUploadPictureInterfaceHandler implements DisposableBean, Runnable {

    @Value("${electronic.ElectronicCoupletUploadPictureController.Topic}")
    private String topic;

    private Thread thread;

    @Autowired
    private BaseStringKafkaConsumer baseStringKafkaConsumer;
    @Autowired
    private ElectronicCoupletUploadPictureService electronicCoupletUploadPictureService;
    private boolean flag;

    @Value("${interface.isTest}")
    private String isTest;

    @Autowired
    public void initThread() {
        if ("true".equals(isTest)) {
            return;
        }
        this.flag = true;
        this.thread = new Thread(this);
        thread.setName("ElectronicCoupletUploadPictureInterfaceHandlerThread");
        this.thread.start();
        log.info("ElectronicCoupletUploadPictureInterfaceHandler start");
    }

    @Override
    public void run() {
        try {
            // 消费前置机KAFKA
            KafkaConsumer<String, String> consumer = baseStringKafkaConsumer.getConsumer(topic);
            while (flag) {
                boolean success = true;
                ConsumerRecords<String, String> messages = consumer.poll(Duration.ofMillis(100L));
                List<ElectronicCoupletUploadPicture> results = new ArrayList<>();
                for (ConsumerRecord<String, String> message : messages) {
                    List<ElectronicCoupletUploadPicture> entry = JSONObject
                            .parseArray(message.value(), ElectronicCoupletUploadPicture.class);
                    if(entry.size()>0){
                        results.add(entry.get(0));
                    }
                }
                //写入内置机KAFKA
                if (!CollectionUtils.isEmpty(results)) {
                    for (ElectronicCoupletUploadPicture elecInfo : results) {
                        try {
                            electronicCoupletUploadPictureService.save(elecInfo);
                            consumer.commitSync();
                        } catch (Exception e) {
                            if (e.getMessage().contains("Duplicate entry"))
                                consumer.commitSync();
                            log.error("ElectronicCoupletUploadPictureInterfaceHandler commit offset failed {}",
                                    e.getMessage());
                            e.printStackTrace();
                            //success = false;
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("ElectronicCoupletUploadPictureInterfaceHandler run error");
            e.printStackTrace();
            //success = false;
        }
    }

    @Override
    public void destroy() throws Exception {
        log.info("ElectronicCoupletUploadPictureInterfaceHandlerThread is down");
        this.flag = false;
    }
}
